package com.atguigu.day09;

import com.atguigu.utils.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

// 使用sql实现实时热门商品
public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<UserBehavior> stream = env
                .readTextFile("/home/zuoyuan/flink0609/src/main/resources/UserBehavior.csv")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] array = value.split(",");
                        return new UserBehavior(
                                array[0],
                                array[1],
                                array[2],
                                array[3],
                                Long.parseLong(array[4]) * 1000L
                        );
                    }
                })
                .filter(r -> r.type.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, settings);

        Table table = streamTableEnvironment
                .fromDataStream(
                        stream,
                        $("userId"),
                        $("itemId"),
                        $("categoryId"),
                        $("type"),
                        $("ts").rowtime() // 将ts指定为事件时间
                );

        // 注册成临时视图
        streamTableEnvironment
                .createTemporaryView("clicks", table);

        // 滑动窗口：HOP(使用的时间戳，滑动距离，窗口长度)
        // 滚动窗口：TUMBLE(使用的时间戳，窗口长度)
        // 第一步：计算ItemViewCountPerWindow
        String innerSQL = "SELECT itemId, COUNT(itemId) as itemCount, " +
                        "HOP_START(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) as windowStart, " +
                        "HOP_END(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) as windowEnd " +
                        "FROM clicks GROUP BY " +
                        "itemId, HOP(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS)";

        // 第二步：使用windowEnd进行分区，按照itemCount字段进行降序排列，ROW_NUMBER()是行号也就是排名
        String midSQL = "SELECT *, ROW_NUMBER() OVER (PARTITION BY windowEnd ORDER BY itemCount DESC) as row_num" +
                " FROM (" + innerSQL + ")";

        // 第三步：取出前三名
        String outerSQL = "SELECT * FROM (" + midSQL + ") WHERE row_num <= 3";

        Table result = streamTableEnvironment
                .sqlQuery(outerSQL);

        streamTableEnvironment
                .toChangelogStream(result)
                .print();

        env.execute();
    }
}
